//
// Created by 33145 on 2023/5/28.
//

#include <sys/socket.h>
#include <sys/epoll.h>
#include <cstring>
#include <sys/eventfd.h>
#include "../common/log.h"
#include "../common/util.h"
#include "eventloop.h"

#define ADD_TO_EPOLL(event) \
  auto it = _listen_fds.find(event->getFd()); \
  int op = EPOLL_CTL_ADD; \
  if (it != _listen_fds.end()) { \
    op = EPOLL_CTL_MOD;   \
  }  \
  epoll_event tmp = event->getEpollEvent();  \
  int rt = epoll_ctl(_epoll_fd, op, event->getFd(), &tmp);  \
  if (rt  == -1) {  \
    LOG_ERROR("addEpollEvent error : %s", strerror(errno));  \
    exit(0);  \
  }                         \
  _listen_fds.insert(event->getFd());

#define DEL_TO_EPOLL(event) \
  auto it = _listen_fds.find(event->getFd()); \
  int op = EPOLL_CTL_DEL; \
  if (it == _listen_fds.end()) {  \
      return;  \
  }  \
  int rt = epoll_ctl(_epoll_fd, op, event->getFd(), nullptr);  \
  if (rt == -1) {  \
    LOG_ERROR("delEpollEvent error : %s", strerror(errno));  \
    exit(0);  \
  }                         \
  _listen_fds.erase(event->getFd());

namespace rocket {

static thread_local EventLoop::ptr t_current_eventloop = nullptr;
static const int g_epoll_max_timeout = 10000;
static const int g_epoll_max_events = 5000;

  EventLoop::EventLoop() {
    if (t_current_eventloop != nullptr) {
      LOG_ERROR("multi eventloop!");
      exit(0);
    }
    _thread_id = getThreadId();
    _epoll_fd = epoll_create(5000);
    if (_epoll_fd  == -1) {
      LOG_ERROR("epoll_creat error : %s", strerror(errno));
      exit(0);
    }

    LOG_INFO("create eventloop in thread : %d", _thread_id);

    initWakeUpFdEvent();
    initTimer();

    // t_current_eventloop = shared_from_this();
  }
  EventLoop::~EventLoop() {
    close(_epoll_fd);
    if (_wakeup_fd_event) {
      delete _wakeup_fd_event;
      _wakeup_fd_event = nullptr;
    }
  }

  void EventLoop::initWakeUpFdEvent() {
    _wakeup_fd = eventfd(0, EFD_NONBLOCK);
    if (_wakeup_fd < 0) {
      LOG_ERROR("wakeup_fd create error : %s", strerror(errno));
      exit(0);
    }
    _wakeup_fd_event = new WakeUpFdEvent(_wakeup_fd);
    _wakeup_fd_event->listen(FdEvent::IN_EVENT, [this](){  // wakeupfd被唤醒时要把所有的内容都读完
      char buf[8];
      while(read(_wakeup_fd, buf, 8) != -1 && errno != EAGAIN);
      LOG_DEBUG("onRead full bytes from wakeup_fd : %d", _wakeup_fd);
    });
    // _listen_fds.insert(_wakeup_fd);
    addEpollEvent(_wakeup_fd_event);
    _listen_fds.insert(_wakeup_fd);
  }

  void EventLoop::loop() {
    _is_looping = true;
    while(!_stop_flag) {
      ScopeMutex<Mutex> lock(_mutex);
      std::queue<std::function<void()>> tmp_tasks;
      _pending_tasks.swap(tmp_tasks);   // TODO 他一次就执行完所有的任务吗？感觉还是一次执行一个任务好一点吧
      lock.unlock();

      while(!tmp_tasks.empty()) {
        auto task = tmp_tasks.front();
        tmp_tasks.pop();
        if (task != nullptr) {
          task();
        }
      }

      int timeout = g_epoll_max_timeout;  // 和sylar的外挂版定时器不同 他使用了timerfd套接字，到时就会提醒
                                          // sylar是通过设置epoll_wait来实现定时器效果
      epoll_event result_events[g_epoll_max_events];
      LOG_DEBUG("prepare to epoll_wait \n");
      int rt = epoll_wait(_epoll_fd, result_events, g_epoll_max_events, timeout);
      if (rt < 0) {
        LOG_ERROR("epoll_wait error : %s", strerror(errno));
        exit(0);
      }
      for (int i = 0; i < rt; i++) {
        // printf("wakeup from epoll_wait!\n");
        epoll_event trigger_event = result_events[i];
        FdEvent* fd_event = static_cast<FdEvent*>(trigger_event.data.ptr);
        if (fd_event && (trigger_event.events & EPOLLIN)) {
          addTask(fd_event->handler(FdEvent::IN_EVENT));
        }
        if (fd_event && (trigger_event.events & EPOLLOUT)) {
          addTask(fd_event->handler(FdEvent::OUT_EVENT));
        }

        if (trigger_event.events & EPOLLERR) {
          LOG_DEBUG("fd %d trigger EPOLLERROR event", fd_event->getFd())
          // 删除出错的套接字
          delEpollEvent(fd_event);
          if (fd_event->handler(FdEvent::ERROR_EVENT) != nullptr) {
            LOG_DEBUG("fd %d add error callback", fd_event->getFd());
            addTask(fd_event->handler(FdEvent::OUT_EVENT));
          }
        }
      }
    } // big while end
  }

  void EventLoop::addEpollEvent(FdEvent* event) {
    if (isInLoopThread()) {
      auto it = _listen_fds.find(event->getFd());
      int op = EPOLL_CTL_ADD;
      if (it != _listen_fds.end()) {
        op = EPOLL_CTL_MOD;
      }
      epoll_event tmp = event->getEpollEvent();
      int rt = epoll_ctl(_epoll_fd, op, event->getFd(), &tmp);
      if (rt  == -1) {
        // printf("addEpollEvent error : %s\n", strerror(errno));
        LOG_ERROR("addEpollEvent error : %s", strerror(errno));
        exit(0);
      }
      _listen_fds.insert(event->getFd());
    } else {
      addTask([this, event](){
        ADD_TO_EPOLL(event);
      }, true);
    }
  }

  void EventLoop::delEpollEvent(FdEvent* event) {
    if (isInLoopThread()) {
      DEL_TO_EPOLL(event);
    } else {
      addTask([this, event](){
        DEL_TO_EPOLL(event);
      }, true);
    }
  }

  void EventLoop::addTask(std::function<void()> cb, bool is_wake_up /*false*/) {
    ScopeMutex<Mutex> lock(_mutex);
    _pending_tasks.push(cb);
    if (is_wake_up) {
      wakeUp();
    }
  }

  bool EventLoop::isInLoopThread() const {
    return getThreadId() == _thread_id;
  }

  void EventLoop::dealWakeUp() {

  }

  void EventLoop::wakeUp() {
    if (_wakeup_fd_event != nullptr) {
      _wakeup_fd_event->wakeup();
    }

  }
  void EventLoop::stop() {
    _stop_flag = true;
    wakeUp();
  }

  void EventLoop::initTimer() {
    _timer = std::make_shared<Timer>();
    addEpollEvent(_timer.get());
  }

  void EventLoop::addTimerEvent(TimerEvent::ptr event) {
      _timer->addTimerEvent(event);
  }

  EventLoop::ptr EventLoop::GetCurrentEventLoop() {
    if (t_current_eventloop) {
      return t_current_eventloop;
    }
    t_current_eventloop = std::make_shared<EventLoop>();
    return t_current_eventloop;
  }
}